package com.example.canal.strategy.canal;

import com.alibaba.fastjson.JSONObject;
import com.example.canal.common.bean.CanalBean;
import com.example.canal.strategy.config.DbChangeProcessor;
import com.example.canal.strategy.service.AbstractDBChangeService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;


/**
 * @author LiYuhang
 * @version 0.1
 * @application
 * @Date 2021/8/8 18:26
 */
@Component
@Slf4j
public final class CanalStrategyConsumer {


    /**
     * 监听的队列名称为：canaltopic
     *
     * @param consumer
     * @param ack
     */
    @KafkaListener(topics = "canaltopic", groupId = "topic.canal")
    public void receive(ConsumerRecord<?, ?> consumer,Acknowledgment ack) {
        String value = (String) consumer.value();
        log.info("topic名称:{},key:{},分区位置:{},下标:{},value:{}",consumer.topic(),consumer.key(),consumer.partition(),consumer.offset(),value);
        //转换为javaBean
        CanalBean canalBean = JSONObject.parseObject(value,CanalBean.class);
        processData(canalBean);
        ack.acknowledge();
    }

    /**
     * 处理数据
     *
     * @param canalBean
     */
    private void processData(CanalBean canalBean) {
        //获取是否是DDL语句
        boolean isDdl = canalBean.isDdl();
        String tableName = canalBean.getTable();
        if (isDdl || StringUtils.isBlank(tableName)) {
            return;
        }
        AbstractDBChangeService service = DbChangeProcessor.getDBChangeService(tableName);
        if (service != null) {
            service.process(canalBean);
        }
    }


}
